package com.yanfan.mq.redischannel.consumer;

import com.alibaba.fastjson2.JSONObject;
import com.yanfan.base.service.ISessionStore;
import com.yanfan.common.constant.YanfanConstant;
import com.yanfan.common.core.device.DeviceAndProtocol;
import com.yanfan.common.core.mq.DeviceStatusBo;
import com.yanfan.common.core.mq.MQSendMessageBo;
import com.yanfan.common.core.thingsModel.ThingsModelSimpleItem;
import com.yanfan.common.enums.DeviceStatus;
import com.yanfan.iot.cache.IDeviceCache;
import com.yanfan.iot.domain.Device;
import com.yanfan.iot.domain.Product;
import com.yanfan.iot.model.ThingsModels.ThingsModelShadow;
import com.yanfan.iot.service.IDeviceService;
import com.yanfan.iot.service.IProductService;
import com.yanfan.iot.util.SnowflakeIdWorker;
import com.yanfan.mq.model.ReportDataBo;
import com.yanfan.mq.redischannel.producer.MessageProducer;
import com.yanfan.mq.service.IMqttMessagePublish;
import com.yanfan.mq.service.IRuleEngine;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;

import javax.annotation.Resource;
import java.util.List;

/**
 * 设备状态消息处理类
 *
 * @author gsb
 * @date 2022/10/10 11:02
 */
@Slf4j
@Component
public class DeviceStatusConsumer {

    @Autowired
    private IDeviceCache deviceCache;
    @Resource
    private IRuleEngine ruleEngine;
    @Resource
    private IDeviceService deviceService;
    @Resource
    private IProductService productService;
    @Resource
    private ISessionStore sessionStore;
    @Resource
    private IMqttMessagePublish mqttMessagePublish;
    @Value("${server.broker.enabled}")
    private Boolean enabled;
    private final SnowflakeIdWorker snowflakeIdWorker = new SnowflakeIdWorker(2);

    public synchronized void consume(DeviceStatusBo bo) {
        try {
            Device device = deviceService.selectDeviceBySerialNumber(bo.getSerialNumber());
            Product product = productService.selectProductByProductId(device.getProductId());
            if (enabled && product.getDeviceType() != 3) { //如果使用Netty版本 监控设备不走这里
                boolean containsKey = sessionStore.containsKey(bo.getSerialNumber());
                boolean isOnline = device.getStatus() == 3;
                log.info("=>session：{},数据库：{}，更新状态:{}", containsKey, isOnline, bo.getStatus().getCode());
                if (containsKey && !isOnline) {
                    //如果session存在，但数据库状态不在线，则以session为准
                    bo.setStatus(DeviceStatus.ONLINE);
                }
                if (!containsKey && isOnline) {
                    bo.setStatus(DeviceStatus.OFFLINE);
                }
            }
            /*更新设备状态*/
            deviceCache.updateDeviceStatusCache(bo, device);
            //处理影子模式值
            this.handlerShadow(device, bo.getStatus());
            //设备上下线执行规则引擎
            ReportDataBo dataBo = new ReportDataBo();
            dataBo.setRuleEngine(true);
            dataBo.setProductId(device.getProductId());
            dataBo.setType(bo.getStatus().equals(DeviceStatus.ONLINE) ? 5 : 6);
            dataBo.setSerialNumber(bo.getSerialNumber());
            ruleEngine.ruleMatch(dataBo);
        } catch (Exception e) {
            log.error("=>设备状态处理异常", e);
        }
    }

    private void handlerShadow(Device device, DeviceStatus status) {
        //获取设备协议编码
        DeviceAndProtocol dp = deviceService.selectProtocolBySerialNumber(device.getSerialNumber());
        String protocolCode = dp.getProtocolCode();
        /* 设备上线 处理影子值*/
        if (status.equals(DeviceStatus.ONLINE) && device.getIsShadow() == 1) {
            ThingsModelShadow shadow = deviceService.getDeviceShadowThingsModel(device);
            List<ThingsModelSimpleItem> properties = shadow.getProperties();
            List<ThingsModelSimpleItem> functions = shadow.getFunctions();
            //JsonArray组合发送
            if (YanfanConstant.PROTOCOL.JsonArray.equals(protocolCode)) {
                if (!CollectionUtils.isEmpty(properties)) {
                    mqttMessagePublish.publishProperty(device.getProductId(), device.getSerialNumber(), properties, 3);
                }
                if (!CollectionUtils.isEmpty(functions)) {
                    mqttMessagePublish.publishFunction(device.getProductId(), device.getSerialNumber(), functions, 3);
                }
            } else { //其他协议单个发送
                functions.addAll(properties);
                if (!CollectionUtils.isEmpty(functions)) {
                    for (ThingsModelSimpleItem function : functions) {
                        MQSendMessageBo bo = new MQSendMessageBo();
                        bo.setDp(dp);
                        bo.setShadow(false);
                        bo.setIdentifier(function.getId());
                        bo.setSerialNumber(device.getSerialNumber());
                        JSONObject jsonObject = new JSONObject();
                        jsonObject.put(function.getId(), function.getValue());
                        bo.setParams(jsonObject);
                        bo.setValue(function.getValue());
                        long id = snowflakeIdWorker.nextId();
                        bo.setMessageId(id + "");
                        //发送到MQ处理
                        MessageProducer.sendFunctionInvoke(bo);
                    }
                }
            }
        }
    }
}
